import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.BitSet;
import java.util.LinkedList;

/**
 * Handles the interaction with one peer,
 * including input and output of messages.
 * Uses one thread to manage input stream and one thread for output.
 * Carries out actions associated with messages.
 * 
 * @author io_porges
 *
 */
public class PeerHandler extends Thread implements Comparable<PeerHandler> {
	private PeerInfo info; //contains id, ip, port
	private Socket s;
	private InputStream in;
	private PeerOutputStream out;
	private BitSet remoteSet;
	private LinkedList<Request> outstandingRequests;

	private GlobalTorrentInfo gTorrentInfo;
	private boolean receivedHandshake, sentHandshake, canTakeBitfield;
	private boolean peerChokingMe, amChokingPeer, amInterested;
	private boolean terminatePeer;
	private boolean outgoing;
	private long timeSinceLastUnchoke;

	public static final int peerTimeout = 60; // in seconds

	public String toString() {
		if (receivedHandshake)
			return '*' + info.address();
		return info.address();
	}


	/**
	 * pre-condition: this is a peer who wants to download our torrent
	 * 					Socket sent_socket is already authenticated when 
	 * 					PeerConnectionAccepter established connection.
	 * 
	 * @param sent_socket
	 * @param peer
	 */
	private PeerHandler (GlobalTorrentInfo g) {
		info = null;
		s = null;
		in = null;
		out = null;
		gTorrentInfo = g;
		receivedHandshake = false;
		sentHandshake = false;
		canTakeBitfield = true;
		amChokingPeer = true;
		amInterested = false;
		peerChokingMe = true;
		terminatePeer = false;
		remoteSet = new BitSet ();
		outstandingRequests = new LinkedList<Request> ();
	}

	public static PeerHandler startIncomingPeer (Socket inSocket, PeerInfo peer, 
			GlobalTorrentInfo info) {
		if (peer == null){
			System.out.println("Attempted to start an incoming connection while peer was null");
		}
		PeerHandler p = new PeerHandler (info);
		p.info = peer;
		p.s = inSocket;
		p.receivedHandshake = true;
		p.outgoing = false;
		p.start();
		return p;
	}

	public static PeerHandler startOutgoingPeer (PeerInfo peer, 
			GlobalTorrentInfo info) {

		PeerHandler p = new PeerHandler (info);
		p.info = peer;
		p.outgoing = true;
		p.start();
		return p;
	}

	public void run () {
		try {
			if (outgoing) {
				if (GlobalTorrentInfo.debugConn)
					System.out.println("attemping outgoing conn: " + info.address());
				try {
					s = new Socket ();
					s.connect(new InetSocketAddress(info.address(),info.port()), 5000);
				} catch (IOException e) {
					if (GlobalTorrentInfo.debugConn)
						System.out.println("Couldn't establish connection with: "+info.address());
					closeConnection();
					return;
				}
			} else if (GlobalTorrentInfo.debugConn)
				System.out.println("received incoming conn:" + info.address());
			in = s.getInputStream();
			out = new PeerOutputStream (s.getOutputStream(), gTorrentInfo);
			if (!sentHandshake) {
				out.handshake(gTorrentInfo);
			}
			if (!receivedHandshake) {
				s.setSoTimeout(5000);
				String peerID = Message.readHandshake(in, gTorrentInfo.torrentFile);
				if (peerID == null) {
					if (GlobalTorrentInfo.debugConn)
						System.out.println("bad return handshake");
					closeConnection();
					return;
				}
				info.setPeerID(peerID);
				receivedHandshake = true;
			}
			if (GlobalTorrentInfo.debugConn)
				System.out.println("handshake complete: " + info.address());
			BitSet bits = gTorrentInfo.dataFileHandler.getValidPieces();
			if (bits.cardinality() > 0)
				out.bitField(bits);
			timeSinceLastUnchoke = System.currentTimeMillis();
		} catch (IOException e) {
			if (GlobalTorrentInfo.debugConn)
				System.out.println("IO Exception. Connection couldn't be est. w/ peer: "+info.address());
			closeConnection();
			return;
		}
		try {
			while (!terminatePeer) {
				handleIncomingMessage();
			}
		} catch (IOException e) {
			if (GlobalTorrentInfo.debugConn)
				System.out.println("error with peer, terminating connection");
			closeConnection();
		} // abnormal behavior, so close connection
	}

	private void handleIncomingMessage () throws IOException {
		Message msg = null;
		try {
			s.setSoTimeout(peerTimeout*666);              // this waits for 1.5 minutes
			msg = Message.readMsg(in);          // and sends keepAlive so
		} catch (SocketTimeoutException e) {    // peer doesn't terminate conn.
			out.keepAlive();                    // then waits for 30 more sec.
			s.setSoTimeout(peerTimeout*333);              // and throws exception
			msg = Message.readMsg(in);          // if no msg received
		}

		if (msg == null) {
			if (GlobalTorrentInfo.debugConn)
				System.out.println(info.address() + " sent bad msg...terminating");
			closeConnection();
			return;
		}

		if (GlobalTorrentInfo.debugConn)
			System.out.println(info.address() + ": " + msg);
		if (msg.isKeepAlive())
			return;

		else if (msg.isBitfield() && canTakeBitfield) {
			//payload represents which pieces the sender has. update remoteSet
			remoteSet = msg.returnBitfield();
			if (GlobalTorrentInfo.debugConn)
				System.out.println(info.address() + " has "+remoteSet.cardinality()+" pieces");
			amInterested = checkIfInterested();
			if (amInterested)
				out.interested();
		}
		else if(msg.isCancel()) {
			//not in design specification, do nothing
		}
		else if(msg.isChoke()) {
			//everything starts as choked. Choke not in design specification 
			peerChokingMe = true;
		}
		else if(msg.isInterested()){
			//peer is interested in what you have. Unchoke peer.
			out.unchoke();
			amChokingPeer = false;
		}           
		else if(msg.isNotInterested()){
			if (!amChokingPeer) {
				amChokingPeer = true;
				out.choke();
			}
		}
		else if(msg.isUnChoke()){
			//Peer has unchoked you. You may begin requesting pieces.
			peerChokingMe = false;
			timeSinceLastUnchoke = System.currentTimeMillis();
		}
		else if(msg.isRequest()){
			//Peer has requested a block. Send the appropriate piece if the peer is unchoked.
			if(!amChokingPeer){
				BlockInfo b = msg.returnBlockInfo();
				byte[] data = null;
				try {
					data = gTorrentInfo.dataFileHandler.read(b);
				} catch (IOException e) {}
				if (data == null) {
					if (GlobalTorrentInfo.debugConn)
						System.out.println("Peer requested bad data");
				} else
					out.piece(b.index(), b.offset(), data);             
			}
			else {
				if (GlobalTorrentInfo.debugConn)
					System.out.println("A choked peer made a request which was ignored");
			}
		}
		else if(msg.isPiece()){
			//Peer has sent a block. Send it to the FileHandler for confirmation and writing.
			byte[] data = msg.returnPieceBytes();
			BlockInfo block = msg.returnBlockInfo();
			synchronized (outstandingRequests) {
				Request tmp = new Request (block);
				for (int k=0; k<outstandingRequests.size(); k++) {
					Request r = outstandingRequests.remove();
					if (!r.equals(tmp))
						outstandingRequests.add(r);
				}
			}
			gTorrentInfo.tracker.increaseDownloadedBy(block.length());
			if (GlobalTorrentInfo.debugPR)
				System.out.println("piece: "+block+" from "+info.address());
			boolean success = gTorrentInfo.dataFileHandler.write(block, data);
			if (GlobalTorrentInfo.debugPR)
				if (success)
					System.out.println("piecewritten: "+block);
				else
					System.out.println("pieceNOTwrit: "+block);
		}
		else if(msg.isHave()){
			//Peer has completed downloading a piece. Send an interested message if appropriate.
			remoteSet.set(msg.returnHave(), true);
			if (GlobalTorrentInfo.debugConn)
				System.out.println(info.address() + " has "+remoteSet.cardinality()+" pieces");
			if (!amInterested) {
				amInterested = checkIfInterested(msg.returnHave());
				if (amInterested)
					out.interested();
			}
		}

		//TODO : when a piece completes, check the SHA-1 hash and send a have message to all peers
		//TODO: when the file completes, communicate with the torrent. check the SHA-1 hash of the entire file.

		canTakeBitfield = false;
		if (!peerChokingMe)
			timeSinceLastUnchoke = System.currentTimeMillis();
		if (peerChokingMe && ((System.currentTimeMillis()-timeSinceLastUnchoke)>peerTimeout*1000))
			closeConnection();
	}


	public void closeConnection() {
		if (terminatePeer)
			return; // connection already terminated
		terminatePeer = true;
		this.interrupt();
		try {
			s.close();
		} catch (IOException e) {}
		gTorrentInfo.peers.removePeerConnection(this);
	}







	/**
	 * Checks if you are interested in the blocks this peer has (represented in remoteSet)
	 * If so, sends an interested message for the next block.
	 * 
	 * @return
	 * @throws IOException 
	 */
	private boolean checkIfInterested () {
		BitSet b = (BitSet) remoteSet.clone();
		b.andNot(gTorrentInfo.dataFileHandler.getValidPieces());

		return (b.cardinality() > 0);
	}

	private boolean checkIfInterested (int pieceIndex) {
		if (pieceIndex >= 0)
			return !gTorrentInfo.dataFileHandler.completedPiece(pieceIndex);
		return checkIfInterested();
	}

	public boolean equals (Object obj) {
		if (!(obj instanceof PeerHandler))
			return false;
		return info.equals(((PeerHandler) obj).info);
	}

	public boolean connected () {
		return receivedHandshake;
	}

	public boolean isChoked() {
		return peerChokingMe;
	}

	public int compareTo(PeerHandler o) {
		if (o == null)
			return 1;
		return info.compareTo(o.info);
	}

	public PeerInfo peerInfo() {
		return info;
	}

	public void sendHave (int pieceIndex) throws IOException {
		amInterested = checkIfInterested();
		if (out != null) {
			out.have(pieceIndex);
			if (!amInterested)
				out.notInterested();
		}
	}

	public void fillRequests () throws IOException {
		if (!peerChokingMe)
			synchronized (outstandingRequests) {
				for (int k=0; k<outstandingRequests.size(); k++) {
					Request r = outstandingRequests.remove();
					if (!r.timedOut())
						outstandingRequests.add(r);
				}
				try {
					while (amInterested && !peerChokingMe && outstandingRequests.size() < PeersInterface.maxRequests)
						requestNewPiece();
				} catch (ArrayIndexOutOfBoundsException e) {return;}
			}
	}

	public void requestNewPiece() throws IOException {
		if (!peerChokingMe)
			synchronized (outstandingRequests) {
				if (outstandingRequests.size() < PeersInterface.maxRequests) {
					BlockInfo b = gTorrentInfo.dataFileHandler.nextBlock(remoteSet);
					if (b == null)
						return;
					out.request(b.index(), b.offset(), b.length());
					if (GlobalTorrentInfo.debugPR)
						System.out.println("request: "+b+" to "+info.address());
					outstandingRequests.add(new Request(b));
				}
			}
	}

	public static class Request implements Comparable<Request> {
		public static final long timeout = peerTimeout*1000;  // 30 seconds
		public BlockInfo block;
		private long time;

		private Request (BlockInfo block) {
			this.block = block;
			this.time = System.currentTimeMillis();
		}

		public boolean timedOut () {
			return System.currentTimeMillis() > timeout + time;
		}

		public int compareTo (Request r) {
			int index = r.block.index() - block.index();
			if (index != 0)
				return index;
			int offset = r.block.offset() - block.offset();
			if (offset != 0)
				return offset;
			return r.block.offset() - block.offset();
		}

		public String toString () {
			return block + " "+ (System.currentTimeMillis()-time);
		}

		public boolean equals (Object obj) {
			if (!(obj instanceof Request))
				return false;
			return compareTo ((Request) obj) == 0;
		}

	}
}